package com.kl.example.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * 生产者拦截器示例
 * 可以指定多个拦截器形成拦截器链
 */
public class ProducerInterceptorExample implements ProducerInterceptor<String, String> {

    private long sendSuccess = 0;
    private long sendFailed = 0;

    /**
     * 在消息发送后进行拦截
     */
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
        // 将发送的消息拦截后加上一个前缀
        String modifyValue = "Prefix-" + producerRecord.value();
        return new ProducerRecord<>(producerRecord.topic(), producerRecord.partition(),
                producerRecord.timestamp(), producerRecord.key(),
                modifyValue,
                producerRecord.headers());
    }

    /**
     * RecordMetadata 和 RecordMetadata 参数是互斥的
     * 如果出现异常那么RecordMetadata为空
     * 如果没有出现异常，那么RecordMetadata有值
     */
    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        // 记录消息发送的成功次数和失败次数
        if (e == null) {
            sendSuccess++;
        } else {
            sendFailed++;
        }
    }

    @Override
    public void close() {
        double successRatio = (double) sendSuccess / (sendSuccess + sendFailed);
        // %f表示以十进制格式化输出浮点数
        System.out.printf("消息发送的成功率 = %f%s%n", successRatio * 100, "%");
    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}
